package com.atguigu.wordcount;


import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class WC1 {

    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 2. 从文件读取数据  按行读取(存储的元素就是每行的文本)
        DataSource<String> ds = env.readTextFile("input/words.txt");

        // 3. 转换数据格式
        FlatMapOperator<String, String> stringStringFlatMapOperator = ds.flatMap((FlatMapFunction<String, String>) (line, collector) -> {
            String[] words = line.split(" ");
            for (String word : words) {
                collector.collect(word);
            }
        }).returns(Types.STRING);

        // 4. 按照 word 进行分组
        MapOperator<String, Tuple2<String, Long>> map = stringStringFlatMapOperator.map(word -> Tuple2.of(word, 1L)).returns(Types.TUPLE(Types.STRING, Types.LONG));

        // 5. 分组内聚合统计
        UnsortedGrouping<Tuple2<String, Long>> tuple2UnsortedGrouping = map.groupBy(0);

        AggregateOperator<Tuple2<String, Long>> sum = tuple2UnsortedGrouping.sum(1);
        // 6. 打印结果

        sum.print();


    }



}
